package com.company.nuwa.redis.support.ext;

import cn.hutool.extra.spring.SpringUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RFuture;
import org.redisson.api.RedissonClient;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;

/**
 * 基于Redis实现延时队列{@link RDelayedQueue}
 *
 * @param <T>
 * @author Nikola Tesla
 * @version 1.0
 * @date 2024/05/31 13:42
 *
 * <pre>
 *
 *     // 发送延时消息
 *     public void publishDelayQueue() {
 *         RedisDelayQueue<String> delayQueue = RedisDelayQueue.newInstance("test-delay-message");
 *
 *         delayQueue.publish("this is a test 1 100s, content:" + RandomUtil.randomString(6), 100, TimeUnit.SECONDS);
 *         delayQueue.publish("this is a test 2 10s, content:" + RandomUtil.randomString(6), 10, TimeUnit.SECONDS);
 *     }
 *
 *     // 初始化消费者
 *     {@code @PostConstruct }
 *     public void initConsumerDelayQueue() {
 *         RedisDelayQueue<String> delayQueue = RedisDelayQueue.newInstance("test-delay-message");
 *         delayQueue.consumer(new Consumer<String>() {
 *             {@code @Override }
 *             public void accept(String message) {
 *                 log.info("accept message:{}", message);
 *             }
 *         });
 *     }
 * </pre>
 */
@Slf4j
public class RedisDelayQueue<T> {

    private final String name;

    private final RBlockingDeque<T> blockingDeque;

    private final RDelayedQueue<T> delayedQueue;

    private static final Map<String, RedisDelayQueue> INSTANCE_MAP = new ConcurrentHashMap<>();

    private RedisDelayQueue(String name, RBlockingDeque<T> blockingDeque, RDelayedQueue<T> delayedQueue) {
        this.name = name;
        this.blockingDeque = blockingDeque;
        this.delayedQueue = delayedQueue;
    }

    /**
     * 构造延时队列
     *
     * @param name redis key name
     * @param <T>
     * @return
     */
    public static synchronized <T> RedisDelayQueue<T> newInstance(String name) {
        return newInstance(name, SpringUtil.getBean(RedissonClient.class));
    }

    /**
     * 构造延时队列
     *
     * @param name           redis key name
     * @param redissonClient redisson client
     * @param <T>
     * @return
     */
    public static synchronized <T> RedisDelayQueue<T> newInstance(String name, RedissonClient redissonClient) {
        return INSTANCE_MAP.computeIfAbsent(name, new Function<String, RedisDelayQueue<T>>() {
            @Override
            public RedisDelayQueue<T> apply(String name) {
                // 目标队列
                RBlockingDeque<T> blockingDeque = redissonClient.getBlockingDeque(name);
                // 原生队列
                RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
                return new RedisDelayQueue<>(name, blockingDeque, delayedQueue);
            }
        });
    }

    /**
     * 消费消息
     *
     * @param consumer
     */
    public void consumer(Consumer<T> consumer) {
        Executors.newSingleThreadExecutor().execute(() -> {
            while (true) {
                try {
                    T take = take();
                    consumer.accept(take);
                } catch (Exception e) {
                    log.error("consumer error", e);
                }
            }
        });
    }

    /**
     * 发送延时消息
     *
     * @param message
     * @param delay
     * @param timeUnit
     */
    public void publish(T message, long delay, TimeUnit timeUnit) {
        this.offer(message, delay, timeUnit);
    }

    /**
     * Retrieves and removes the head of the queue represented by this deque
     * (in other words, the first element of this deque), waiting if
     * necessary until an element becomes available.
     *
     * @return the head of this deque
     */
    @SneakyThrows
    public T take() {
        T result = blockingDeque.take();
        log.info("{}[{}] take result={}", this.getClass().getSimpleName(), name, result);
        return result;
    }

    /**
     * Removes all instance of the specified element
     *
     * @param o
     * @return
     */
    public boolean remove(T o) {
        boolean result = delayedQueue.removeIf(t -> delayedQueue.contains(o));
        log.info("{}[{}] remove result={}", this.getClass().getSimpleName(), name, result);
        return result;
    }

    /**
     * Remove all match element
     *
     * @param filter
     * @return
     */
    public boolean removeIf(Predicate<T> filter) {
        boolean result = delayedQueue.removeIf(filter);
        log.info("{}[{}] remove result={}", this.getClass().getSimpleName(), name, result);
        return result;
    }

    /**
     * get all element
     *
     * @return
     */
    public List<T> getAll() {
        return delayedQueue.readAll();
    }

    /**
     * Inserts element into this queue with
     * specified transfer delay to destination queue.
     *
     * @param t        the element to add
     * @param delay    for transition
     * @param timeUnit for delay
     */
    public void offer(T t, long delay, TimeUnit timeUnit) {
        this.delayedQueue.offer(t, delay, timeUnit);
        log.info("{}[{}] offer object={} delay={}, timeUnit={}", this.getClass().getSimpleName(), name, t, delay, timeUnit);
    }

    /**
     * Inserts element into this queue with
     * specified transfer delay to destination queue.
     *
     * @param t        the element to add
     * @param delay    for transition
     * @param timeUnit for delay
     */
    public RFuture<Void> offerAsync(T t, long delay, TimeUnit timeUnit) {
        RFuture<Void> future = this.delayedQueue.offerAsync(t, delay, timeUnit);
        log.info("{}[{}] offerAsync object={} delay={}, timeUnit={} result={}", this.getClass().getSimpleName(), name, t, delay, timeUnit, future);
        return future;
    }

}
